package com.gome.ocean.service.datax.impl;

import java.util.Date;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.ibatis.session.RowBounds;
import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.github.pagehelper.Page;
import com.gome.ocean.common.constants.Constants;
import com.gome.ocean.common.enums.State;
import com.gome.ocean.common.exception.JobCreateException;
import com.gome.ocean.common.exception.JobRunException;
import com.gome.ocean.common.exception.JobServiceException;
import com.gome.ocean.dao.datax.JobDAO;
import com.gome.ocean.dao.model.datax.CreateJobVO;
import com.gome.ocean.dao.model.datax.JobBO;
import com.gome.ocean.service.datax.JobService;
import com.gome.ocean.service.scheduler.SchedulerServiceBean;
import com.gome.ocean.service.task.OneWayTask;
import com.gome.ocean.service.task.TransferJobTask;
import com.gome.ocean.service.thread.ThreadFactory;
import com.gome.ocean.service.velocity.VelocityBean;

/**
 * Created by liuqingxia on 2016/1/21.
 */
@Service
@Transactional("transactionManager")
public class JobServiceImpl implements JobService {

    private final Logger         LOG = LoggerFactory.getLogger(JobServiceImpl.class);

    @Autowired
    private JobDAO               jobDAO;

    @Autowired
    private VelocityBean         velocityBean;

    @Autowired
    private SchedulerServiceBean schedulerServiceBean;

    @Autowired
    private ThreadFactory        threadFactory;

    @Override
    public boolean insertJob(CreateJobVO createJobVO) throws SchedulerException {
        JobBO jobBO = new JobBO();
        if (createJobVO != null) {
            if (createJobVO.getReader() == null || createJobVO.getWriter() == null) {
                throw new JobCreateException("reader or writer 任务参数不能为空!请检查!");
            }
            String pluginData = velocityBean.assemblePlugins(createJobVO.getReader(),
                createJobVO.getReaderMaps(),
                createJobVO.getWriter(),
                createJobVO.getWriterMaps(),
                createJobVO.getJobLimit(),
                createJobVO.getLimitValue());
            jobBO.setContext(pluginData);
            jobBO.setSubmitUser(createJobVO.getSubmit_user());
            jobBO.setSubmitTime(new Date());
            jobBO.setSubmitIp(createJobVO.getSubmit_ip());
            jobBO.setPipelineName(createJobVO.getProject());
            jobBO.setProject(createJobVO.getProject());
            jobBO.setState(State.INIT.getValue());
            jobBO.setProjectId(createJobVO.getProjectId());
            jobBO.setCronExpression(createJobVO.getCronExpression());
            jobBO.setIntroduce(createJobVO.getIntroduce());
            jobBO.setExecuteType(createJobVO.getExecuteType());
            return jobDAO.insertJob(jobBO) > 0;
        }
        return Boolean.FALSE;
    }

    @Override
    public boolean removeJob(JobBO jobBO) throws SchedulerException {
        boolean result = jobDAO.removeJobById(jobBO.getId()) > 0;
        if (result && jobBO.isCronJob()) {
            if (schedulerServiceBean.checkExists(jobBO)) {
                return schedulerServiceBean.removeJob(jobBO);
            }
        }
        return result;
    }

    /**
     * 对于cron类型的job,此方法是向quartz容器中提交job
     *
     * @param jobBO
     * @return
     * @throws SchedulerException
     * @throws JobServiceException
     */
    @Override
    public boolean startJob(JobBO jobBO) throws SchedulerException, JobServiceException {
        if (!jobBO.isCronJob()) {
            throw new JobServiceException("该任务不是cron类型的");
        }
        if (StringUtils.isEmpty(jobBO.getContext())) {
            throw new JobRunException("任务参数不能为空!请检查!");
        }
        if (schedulerServiceBean.checkExists(jobBO)) {
            schedulerServiceBean.removeJob(jobBO);
        }
        return schedulerServiceBean.addJob(jobBO, TransferJobTask.class);
    }

    @Override
    public boolean pauseJob(JobBO jobBO) throws SchedulerException, JobServiceException {
        if (!jobBO.isCronJob()) {
            throw new JobServiceException("该任务不是cron类型的");
        }
        jobBO.setState(State.PAUSE.getValue());
        boolean result = jobDAO.updateJobBO(jobBO) > 0;
        if (result) {
            return schedulerServiceBean.pauseJob(jobBO);
        }
        return Boolean.FALSE;
    }

    @Override
    public boolean resumeJob(JobBO jobBO) throws SchedulerException, JobServiceException {
        if (!jobBO.isCronJob()) {
            throw new JobServiceException("该任务不是cron类型的");
        }
        jobBO.setState(State.RUN.getValue());
        boolean result = jobDAO.updateJobBO(jobBO) > 0;
        if (result) {
            return schedulerServiceBean.resumeJob(jobBO);
        }
        return Boolean.FALSE;
    }

    @Override
    public boolean updateJob(JobBO newJob) throws SchedulerException, JobServiceException {
        JobBO oldJob = jobDAO.findJobById(newJob.getId());
        if (null == oldJob) {
            throw new JobServiceException("您要更新的job不存在");
        }
        boolean result = jobDAO.updateJobBO(newJob) > 0;
        if (result) {
            if (oldJob.isCronJob() && !schedulerServiceBean.checkExists(oldJob)) {
                return Boolean.TRUE;
            }
            if (oldJob.isCronJob() && newJob.isCronJob()) {
                return schedulerServiceBean.rescheduleJob(newJob, TransferJobTask.class);
            }
            if (oldJob.isCronJob() && !newJob.isCronJob()) {
                return schedulerServiceBean.removeJob(oldJob);
            }
            if (!oldJob.isCronJob() && newJob.isCronJob()) {
                return schedulerServiceBean.addJob(newJob, TransferJobTask.class);
            }
        }
        return result;
    }

    @Override
    public boolean triggerJob(JobBO job) throws SchedulerException {
        if (!job.isCronJob()) {
            threadFactory.start(new OneWayTask(job));
            return true;
        }
        return schedulerServiceBean.triggerJob(job);
    }

    @Override
    public boolean interrupt(JobBO jobBO) throws SchedulerException, JobServiceException {
        if (!jobBO.isCronJob()) {
            throw new JobServiceException("该任务不是cron类型的");
        }
        return schedulerServiceBean.interrupt(jobBO);
    }

    @Override
    public JobBO findJob(Long id) {
        return jobDAO.findJobById(id);
    }

    @Override
    public Page<JobBO> findJobsByProjectNames(List<String> projectNames, int pageNum, int pageSize)
                                                                                                   throws JobServiceException {
        Page jobBos;
        try {
            jobBos = (Page) jobDAO.findJobByProjectNames(projectNames, new RowBounds(pageNum, pageSize));
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            throw new JobServiceException(e.toString());
        }
        return jobBos;
    }

    @Override
    public Page<JobBO> findAllJobs(int pageNum, int pageSize) throws JobServiceException {
        Page<JobBO> jobBOs;
        try {
            jobBOs = (Page) jobDAO.findAllJobs(new RowBounds(pageNum, pageSize));
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            throw new JobServiceException(e.getMessage());
        }
        return jobBOs;
    }

    @Override
    public Page<JobBO> findJob(String submitUser, int pageNum, int pageSize) throws JobServiceException {
        Page<JobBO> jobBOs;
        try {
            jobBOs = (Page) jobDAO.findJobBySubmitUser(submitUser, new RowBounds(pageNum, Constants.DEFAULT_PAGE_SIZE));
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            throw new JobServiceException(e.getMessage());
        }
        return jobBOs;
    }

    @Override
    public Page<JobBO> findJob(String submitUser, String projectName, int pageNum, int pageSize)
                                                                                                throws JobServiceException {
        Page<JobBO> jobBOs;
        try {
            jobBOs = (Page) jobDAO.findJobBySubmitUserAndPipelineName(submitUser, projectName, new RowBounds(pageNum,
                pageSize));
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            throw new JobServiceException(e.getMessage());
        }
        return jobBOs;
    }

    @Override
    public Page<JobBO> findJobByProjectName(String projectName, int pageNum, int pageSize) throws JobServiceException {
        Page<JobBO> jobs;
        try {
            jobs = (Page) jobDAO.findJobByPipelineName(projectName, new RowBounds(pageNum, pageSize));
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
            throw new JobServiceException(e.getMessage());
        }
        return jobs;
    }

    @Override
    public Page<JobBO> findJobsByExecuteType(int type, int pageNum, int pageSize) {
        return (Page<JobBO>) jobDAO.findJobByExecuteType(type, new RowBounds(pageNum, pageSize));
    }

    @Override
    public Page<JobBO> findJobsByExecuteTypeAndProjectNames(int type, List<String> projectNames, int pageNum,
                                                            int pageSize) {
        return (Page<JobBO>) jobDAO.findJobsByExecuteTypeAndProjectNames(type, projectNames, new RowBounds(pageNum,
            pageSize));
    }

    @Override
    public Page<JobBO> findJobsByIntroduce(String introduce, int pageNum, int pageSize) {
        return (Page<JobBO>) jobDAO.findJobsByIntroduce(introduce, new RowBounds(pageNum, pageSize));
    }

    @Override
    public Page<JobBO> findJobsByIntroduceAndProject(String introduce, List<String> projectNames, int pageNum,
                                                     int pageSize) {
        return (Page<JobBO>) jobDAO.findJobsByIntroduceAndProject(introduce, projectNames, new RowBounds(pageNum,
            pageSize));
    }

    @Override
    public List<JobBO> findAllJobList() throws JobServiceException {
        return jobDAO.findAllJobList();
    }

    @Override
    public boolean updateJobBO(JobBO jobBO){
        return jobDAO.updateJobBO(jobBO) > 0;
    }

}
